using System;
using System.IO;
using System.Collections.Generic;
using System.Text;
using System.Diagnostics;
using System.Threading;
using System.Net;
using System.Net.Sockets;
using Microsoft.BizTalk.Message.Interop;
using Microsoft.BizTalk.TransportProxy.Interop;

namespace Microsoft.Samples.BizTalk.Adapter.Tcp.Receive
{
    /// <summary>
    ///     This worker thread transmits data from BizTalk to the remote endpoint when the receive port is
    ///     configured as a request-response receive port.
    /// </summary>
    /// <author>
    ///     Paul Brown
    /// </author>
    public class TransmitWorkerThread : WorkerThread
    {
        #region Private Instance Members

        private ReceiveAdapter                          _receiveAdapter;    // The receive adapter

        #endregion

        #region Constructor

        /// <summary>
        ///     This is constructor initializes the instance.
        /// </summary>
        public TransmitWorkerThread(IAdapter adapter)
            : base(adapter)
        {
            // Get thread name prefix
            this.ThreadNamePrefix = TraceHelper.ResourceManager.GetString(ResourceConstants.ReceiveTransmitWorkerThreadName);

            // We know the adapter will be a receive adapter, store a reference to it
            this._receiveAdapter = (ReceiveAdapter)adapter;
        }

        #endregion

        #region Public Instance Properties

        /// <summary>
        ///     This property returns a reference to the receive adapter.
        /// </summary>
        public ReceiveAdapter ReceiveAdapter
        {
            get { return (this._receiveAdapter); }
        }

        #endregion

        #region Private Instance Methods

        /// <summary>
        ///     This method sends the data associated with the work item to the waiting client socket.
        /// </summary>
        private void SendData(TransmitWorkItem workItem)
        {
            //[workitem:6548] 
            // Send data in message
            //SocketIO.SendData(this.Adapter,
            //                  workItem.ClientSocket,
            //                  new Stream[] { workItem.Message.GetPart("body").Data },
            //                  workItem.ReceiveLocation.FrameStart,
            //                  workItem.ReceiveLocation.FrameEnd,
            //                  workItem.ReceiveLocation.DataEncoding,
            //                  workItem.ReceiveLocation.SendDataBufferSize);

            // Send data in message
            SocketIO.SendData(this.Adapter,
                              workItem.ClientSocket,
                              new Stream[] { workItem.Message.BodyPart.Data },
                              workItem.ReceiveLocation.FrameStart,
                              workItem.ReceiveLocation.FrameEnd,
                              workItem.ReceiveLocation.DataEncoding,
                              workItem.ReceiveLocation.SendDataBufferSize);


            // Close socket
            workItem.ClientSocket.Shutdown(SocketShutdown.Send);
            workItem.ClientSocket.Close(workItem.ReceiveLocation.ClientCloseTimeout);
        }

        #endregion

        #region Abstract WorkerThread Members

        #region Public Instance Properties

        /// <summary>
        ///     This property returns the type of a BatchWorkItem which is compatible with this worker thread.
        /// </summary>
        public override Type WorkItemType
        {
            get { return (typeof(TransmitWorkItem)); }
        }

        #endregion

        #region Protected Instance Methods

        /// <summary>
        ///     This method is the delegate that runs the main worker thread.
        /// </summary>
        protected override void Thread()
        {
            // Variables
            bool submitted = false;
            bool                threadInterrupted = false;
            int                 messageCount;
            Stopwatch           timer = new Stopwatch();
            TransmitWorkItem    workItem = null;

            try
            {
                // Loop until we are interrupted
                while (true)
                {
                    // If we have work, start batching process
                    if (this.CurrentWorkLoad > 0)
                    {
                        if (this.Adapter.TraceSwitch.TraceVerbose)
                            TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveGetBatch, this.Name);

                        // Create a new batch for BizTalk
                        Batch batch = new Batch(this.Adapter);
                        batch.RetryCount = this.ReceiveAdapter.BatchRetryCount;
                        batch.RetryInterval = this.ReceiveAdapter.BatchRetryInterval;
                        batch.OrderedBatch = false;
                        messageCount = 0;

                        // Loop while we have work items
                        while (this.CurrentWorkLoad > 0)
                        {
                            if (submitted == true)
                            {
                                TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveInitialiseNewBatch, this.Name);
                                batch = new Batch(this.Adapter);
                                batch.RetryCount = this.ReceiveAdapter.BatchRetryCount;
                                batch.RetryInterval = this.ReceiveAdapter.BatchRetryInterval;
                                batch.OrderedBatch = false;
                                messageCount = 0;
                                submitted = false;    
                            }

                            // Dequeue item
                            workItem = (TransmitWorkItem)this.DequeueWork();

                            if (this.Adapter.TraceSwitch.TraceVerbose)
                                TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveDequeuedTransmitWork, workItem.ReceiveLocation.Uri, this.Name);

                            bool suspendMessage = false;
                            Exception ex = null;
                            try
                            {
                                // Send response
                                this.SendData(workItem);
                            }
                            catch (ThreadInterruptedException e)
                            {
                                // Asked to shut down, set flag and wait until this batch is done
                                threadInterrupted = true;
                                ex = e;
                                suspendMessage = true;
                            }
                            catch (Exception e)
                            {
                                ex = new Exception(TraceHelper.FormatEntry(ResourceConstants.ReceiveFailedToTransmitResponse, workItem.ReceiveLocation.Uri, this.Name, e.Message));
                                TraceHelper.WriteRawEntry(TraceLevel.Error, ex.Message);
                                suspendMessage = true;
                            }

                            // Set up BizTalk batch
                            bool addedSuccessfully = false;
                            while (!addedSuccessfully)
                            {
                                try
                                {
                                    if (suspendMessage)
                                    {
                                        // Suspend message
                                        BatchMessage batchMessage = new BatchMessage(workItem.Message, ex);
                                        batch.MoveToSuspendQ(batchMessage);
                                    }
                                    else
                                    {
                                        // Delete message in BizTalk
                                        BatchMessage batchMessage = new BatchMessage(workItem.Message);
                                        batch.DeleteMessage(batchMessage);
                                    }

                                    addedSuccessfully = true;
                                }
                                catch (ThreadInterruptedException)
                                {
                                    // Shutting down - still retry batch operation
                                    threadInterrupted = true;
                                }
                                catch (Exception e)
                                {
                                    // Failed to add to batch - shouldn't happen in theory so retry
                                    TraceHelper.WriteEntry(TraceLevel.Error, ResourceConstants.ReceiveFailedAddingToBatch, workItem.ReceiveLocation.Uri, batch.BatchCookie, e.Message);
                                }
                            }

                            // Increment controlling properties
                            messageCount++;

                            // Anything in batch?
                            if (messageCount > 0)
                            {
                                bool readyToSubmit = false;

                                // Have we exceeded our batch limits?
                                if (messageCount >= this.ReceiveAdapter.BatchSize)
                                {
                                    if (this.Adapter.TraceSwitch.TraceVerbose)
                                        TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveReachedBatchSizeLimit, this.ReceiveAdapter.BatchSize, messageCount);

                                    readyToSubmit = true;
                                }
                                else if (timer.ElapsedMilliseconds >= this.ReceiveAdapter.BatchTimeout)
                                {
                                    if (this.Adapter.TraceSwitch.TraceVerbose)
                                        TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveExceededBatchTimeoutLimit, this.ReceiveAdapter.BatchTimeout, messageCount);

                                    readyToSubmit = true;
                                }
                                else if (this.CurrentWorkLoad == 0)
                                {
                                    if (this.Adapter.TraceSwitch.TraceVerbose)
                                        TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveSubmittingBatch, messageCount);

                                    readyToSubmit = true;
                                }

                                // Are we ready to submit?
                                if (readyToSubmit)
                                {
                                    // Add to array
                                    if (Batch.GetBatch(batch.BatchCookie) == null)
                                    {
                                        Batch.AddBatch(batch);
                                    }

                                    // Submit batch if limits exceeded or idle
                                    //bool submitted = false;
                                    for (int i = 0; i < this.ReceiveAdapter.BatchRetryCount; i++)
                                    {
                                        try
                                        {
                                            // Submit batch
                                            batch.SubmitBatch();

                                            // Break out of loop, successful
                                            submitted = true;
                                            break;
                                        }
                                        catch (ThreadInterruptedException)
                                        {
                                            // Asked to shut down, set flag and wait until this batch is done
                                            threadInterrupted = true;
                                        }
                                        catch (Exception e)
                                        {
                                            // Trace
                                            TraceHelper.WriteEntry(TraceLevel.Error, ResourceConstants.ReceiveFailedSubmittingBatchRetrying, batch.BatchCookie, e.Message);

                                            // Wait interval
                                            System.Threading.Thread.Sleep(this.ReceiveAdapter.BatchRetryInterval * 1000);
                                        }
                                    }

                                    // If not successful and exhausted all retries
                                    if (!submitted)
                                    {
                                        TraceHelper.WriteEntry(TraceLevel.Error, ResourceConstants.ReceiveFailedSubmittingBatch, batch.BatchCookie);
                                    }
                                }
                            }
                            else
                            {
                                // No messages in batch!
                                if (this.Adapter.TraceSwitch.TraceVerbose)
                                    TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveNoMessagesInBatch);
                            }

                            // Update work item
                            workItem.Status = WorkItemStatusTypes.Complete;
                            workItem = null;
                        }
                    }

                    // Stop timer
                    timer.Stop();

                    // Was the thread interrupted?
                    if (threadInterrupted)
                    {
                        // Throw error again to end thread
                        throw (new ThreadInterruptedException());
                    }

                    if (this.Adapter.TraceSwitch.TraceVerbose)
                        TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.WorkerThreadWaitingForWork, this.Name);

                    // Wait for signal that work has arrived
                    this.WaitForWork();
                }
            }
            catch (ThreadInterruptedException)
            {
                // Asked to shut down
                if (this.Adapter.TraceSwitch.TraceVerbose)
                    TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.WorkerThreadInterrupted, this.Name);
            }
            catch (Exception e)
            {
                TraceHelper.WriteEntry(TraceLevel.Error, ResourceConstants.WorkerThreadGeneralError, this.Name, e.Message);

                // If we have an outstanding work item, push it back onto queue before restarting
                if (workItem != null)
                {
                    this.PushWorkOnQueue(workItem);
                }
            }
            finally
            {
                // Fire terminate event
                this.OnThreadTerminated(new EventArgs());
            }
        }

        #endregion

        #endregion
    }
}
